package com.foogui.boot.controller;

import com.alibaba.fastjson.JSONObject;

import com.foogui.boot.constant.RocketMqConstant;
import com.foogui.boot.constant.RocketMqDelayLevel;
import com.foogui.boot.vo.RocketMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/rocketmq")
@Slf4j
public class RocketMqController {
    @Resource
    RocketMQTemplate rocketMQTemplate;

    /**
     * 通过实体类发送消息，发送注意事项请参考实体类
     * http://localhost:8888/rocketmq/entity/message
     */
    @RequestMapping("/send")
    public Object sendMessage() {
        // 目的：topic:tag，如果不指定则发往配置的默认地址
        String destination = RocketMqConstant.SOURCE_TOPIC + ":" + RocketMqConstant.SOURCE_TAG;
        // 发送的消息是个对象
        RocketMqMessage message = new RocketMqMessage();
        message.setId(System.currentTimeMillis());
        message.setMessage("当前消息发送时间为：" + LocalDateTime.now());
        // Java时间字段需要单独处理，否则会序列化失败
        message.setCurrentDate(LocalDate.now());
        message.setCurrentDateTime(LocalDateTime.now());
        message.setVersion("1.0");
        // 发送为消息头添加业务唯一key
        // 建造者模式创建Message对象
        Message<RocketMqMessage> buildMessage = MessageBuilder.withPayload(message)
                // 设置keys，RocketMQHeaders.KEYS
                .setHeader(RocketMQHeaders.KEYS, message.getId())
                .build();
        // 发送同步消息
        //SendResult sendResult = rocketMQTemplate.syncSend(destination, buildMessage);

        // 发送延迟消息
        //rocketMQTemplate.syncSend(destination, buildMessage, 3000, RocketMqDelayLevel.FIVE_SECOND);

        // 发送同步有序消息，需要指定hashKey，可以用业务唯一键,用于队列的选择
        //rocketMQTemplate.syncSendOrderly(destination, message, message.getId().toString());

        // 调用抽象类方法发送，最终也是syncSend
        rocketMQTemplate.convertAndSend(destination, "convertAndSend");

        // 转换消息和发送，底层使用的是syncSend(destination, message)
        //rocketMQTemplate.send(destination, buildMessage);

        // 发送批量消息
//        List<Message<String>> msgList = new ArrayList<>();
//        for (int i = 0; i < 10; i++) {
//            msgList.add(MessageBuilder.withPayload("消息:" + i).build());
//        }
//        rocketMQTemplate.syncSend(destination, msgList);


        // 发送异步消息，消息发送后及时返回，然后通过回调方法通知
//        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                log.info("消息发送成功【{}】", JSONObject.toJSONString(sendResult));
//            }
//
//            @Override
//            public void onException(Throwable e) {
//                log.error("消息发送失败【{}】", e.getMessage());
//            }
//        });

        // 发送异步有序消息，需要指定hashKey，可以用业务唯一键
        // rocketMQTemplate.asyncSendOrderly(destination, message, message.getId().toString(), new SendCallback() {
        //     @Override
        //     public void onSuccess(SendResult sendResult) {
        //         log.info("消息发送成功【{}】", JSONObject.toJSONString(sendResult));
        //     }
        //
        //     @Override
        //     public void onException(Throwable e) {
        //         log.error("消息发送失败【{}】", e.getMessage());
        //     }
        // });

        // 发送单向消息
        // rocketMQTemplate.sendOneWay(destination, message);

        // 发送单向有序消息，通过MessageBuilder构建
        // rocketMQTemplate.sendOneWayOrderly(destination, buildMessage, message.getId().toString());

        //发送事务消息,Message需要setHeader事务唯一id
        String transactionId = UUID.randomUUID().toString();
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(RocketMqConstant.SOURCE_GROUP, destination,
                MessageBuilder.withPayload(message).setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId).build(), null);
        log.info("发送事务消息（半消息）完成：result = {}", result);

        return "消息发送ok";

    }

    /**
     * 直接将对象进行传输，也可以自己进行json转化后传输
     */
    @RequestMapping("/messageExt/message")
    public SendResult convertAndSend() {
        String destination = "rocketmq_source_code_topic:rocketmq_source_code_ext_tag";
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("type", "messageExt");
        return rocketMQTemplate.syncSend(destination, jsonObject);
    }


}
